#ifndef __LOOPER_H__
#define __LOOPER_H__
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <functional>
#include "Buff.hpp"

namespace ssslog
{
    using Functor = std::function<void(Buffer&)>;
    enum AsyncType
    {
        ASYNC_SAFE,//安全状态表示，缓冲区满了则阻塞
        ASYNC_UNSAFE//非安全状态表示无限扩容，用于性能测试
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor& cb,AsyncType lopper_type = AsyncType::ASYNC_UNSAFE)
                :_stop(false)
                ,_thread(std::thread(&AsyncLooper::ThreadEntry,this))
                ,_callback(cb)
                ,_lopper_type(lopper_type)
                {}
        ~AsyncLooper() { Stop(); }
        void Stop()
        {
            _stop = true;
            _con_cond.notify_all();
            _thread.join();
        }
        void Push(const char* data, size_t size)
        {
            //if(_stop == true) return;
            {
                std::unique_lock<std::mutex> lock(_mtx);
                if(_lopper_type == AsyncType::ASYNC_SAFE)
                    _pro_cond.wait(lock,[&]() { return _pro_buff.WriteBuff() >= size; });
                _pro_buff.Push(data, size);
            }
            _con_cond.notify_all();
        }
    private:
        void ThreadEntry()
        {
            while(1)
            {
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    if(_stop || _pro_buff.Empty()) return;
                    _con_cond.wait(lock,[&]() { return _stop || !_pro_buff.Empty(); });
                    _con_buff.Swap(_pro_buff);
                    if(_lopper_type == AsyncType::ASYNC_SAFE)
                        _pro_cond.notify_all();
                }
                //处理函数
                _callback(_con_buff);
                _con_buff.Reset();
            }
        }
    private:
        Functor _callback;
    private:
        std::atomic<bool> _stop;
        Buffer _pro_buff;//生产者
        Buffer _con_buff;//消费者
        std::condition_variable _pro_cond;//生产者条件变量
        std::condition_variable _con_cond;//消费者条件变量
        std::mutex _mtx;
        std::thread _thread;
        AsyncType _lopper_type;
    };
}
#endif